// -*-mode:c++; coding:utf-8-*-

#include <algorithm>

#include <boost/asio.hpp>
#include <boost/bind.hpp>

#include "logger.hpp"
#include "xbase_utility.hpp"

#include "session.hpp"

namespace xbase
{
	void session::active_async_write()
	{
		boost::mutex::scoped_lock lock(m_mutex);
		while(! m_request_queue.empty())
		{
			packet_ptr packet = m_request_queue.front().first;
			handler_type handler = m_request_queue.front().second;
			lock.unlock(); // 编码过程中不需要持有锁，因为其它调用者只是向队列push，而自己总是pop

			// encode packet
			// TODO: 改进发送缓存的管理方式；现在是逐步尝试
			packet->set_sequence_number(get_packet_id());
			boost::function<bool(protocol::oarchive&)> save_f =
				protocol::get_save_function<protocol::oarchive>(*packet);
			std::size_t buffer_size = protocol::P_BUFFER_SIZE;
			bool ok = false;
			for(;;)
			{
				m_write_buffer.clear();
				protocol::oarchive out(m_write_buffer.prepare(buffer_size));
				ok = save_f(out);
				if(ok)
				{
					m_write_buffer.commit(out.used_buffer_size());
					break;
				}
				else
				{
					buffer_size <<= 1;
					if(buffer_size == 0) // 溢出
						break;
				}
			}
			if(ok)
			{
				boost::asio::async_write(m_socket,
							 m_write_buffer.data(),
							 boost::asio::transfer_all(),
							 boost::bind(&session::on_written,
								     this_ptr(),
								     boost::asio::placeholders::error,
								     boost::asio::placeholders::bytes_transferred));
				break;
				// 这个请求发送完成后，在回调函
				// 数中再发送下一个请求；而不是
				// 一次性将发送队列全部提交。因
				// 为write_buffer只能容纳一个请
				// 求
			}
			
			lock.lock();
			m_request_queue.pop();
			lock.unlock();
			
			// 编码失败，向用户返回失败，继续处理下一个
			this->get_upper_level_service().dispatch(boost::bind(handler,
									     packet_ptr(),
									     error::packet_encode_failed));

			lock.lock();
		}
	}

	void session::on_written(const error_code &ec,
							       std::size_t bytes)
	{
		async_module_stats::bytes_written(bytes);
		if(ec)
		{
			// 发送出错，说明socket出错了；将错误广播，即令
			// 所有待发送的消息失败，并关闭socket

			boost::mutex::scoped_lock lock(m_mutex);
			std::queue<item_type> queue;
			std::swap(queue, m_request_queue);
			lock.unlock();

			LOG(ERROR, "write packet "
			    << (queue.empty()
				? "<NULL>"
				: queue.front().first->packet::to_string())
			    << " error - "
			    << ec << " : " << ec.message() << ", bytes = " << bytes);
			
			while(! queue.empty())
			{
				LOG(ERROR, "cancel request packet " << queue.front().first->to_string());
				this->get_upper_level_service().dispatch(boost::bind(queue.front().second,
										     packet_ptr(),
										     ec));
				queue.pop();
			}
			
			close();
			return;
		}

		boost::mutex::scoped_lock lock(m_mutex);
		item_type item = m_request_queue.front();
		m_request_queue.pop();
		lock.unlock();
		
		async_module_stats::packets_written(1);
		
		// register handler
		const std::pair<handler_map_type::iterator, bool> ret =
			m_handler_map.insert(handler_map_type::value_type
					     (item.first->get_sequence_number(),
					      item.second));
		if(ret.second == false)
		{
			this->get_upper_level_service().dispatch(boost::bind(item.second,
									     packet_ptr(),
									     error::duplicated_id));
		}

		active_async_write();
	}

	void session::active_async_read()
	{
		if(! m_async_read_actived)
		{
			boost::asio::async_read(m_socket,
						m_read_buffer.prepare(protocol::P_BUFFER_SIZE),
						boost::asio::transfer_at_least(protocol::packet::HEADER_SIZE),
						boost::bind(&session::on_read,
							    this_ptr(),
							    boost::asio::placeholders::error,
							    boost::asio::placeholders::bytes_transferred));
			m_async_read_actived = true;
		}
	}

	void session::on_read(const error_code &e,
							    std::size_t bytes_transferred)
	{
		async_module_stats::bytes_read(bytes_transferred);
		m_read_buffer.commit(bytes_transferred);
		std::size_t bytes_to_read = 0;

		error_code ec = e;
		if(! ec)
		{
			std::size_t id = protocol::P_BAD_SEQUENCE;

			for(boost::asio::const_buffer data = m_read_buffer.data();
			    boost::asio::buffer_size(data) != 0;
			    data = m_read_buffer.data())
			{
				const char *begin = boost::asio::buffer_cast<const char*>(data);
				const char *end = begin + boost::asio::buffer_size(data);
				protocol::validate(begin,
						   &end,
						   &bytes_to_read,
						   &id,
						   ec);
				if(ec)
					break;
				
				const std::size_t size = (end - begin);
				protocol::iarchive in(begin, size);
				packet_ptr ptr;
				protocol::new_packet_and_load(ptr, ec, in);
				if(ec)
				{
					const std::size_t max_dump_size = 128;
					LOG(ERROR, "new_packet_and_load failed, bytes_transferred = "
					    << bytes_transferred << ", packet size = "
					    << size << ", packet data = " << dump_string_hex(begin, std::min(size, max_dump_size)));
					dispatch_error(ec);
					close();
					m_async_read_actived = false;
					return;
				}
				async_module_stats::packets_read(1);
				dispatch(ptr, ec);
				m_read_buffer.consume(size);
			}
		}

		if(! ec)	// 没有错误，正好所有数据用完了，则进入下一次循环
		{
			boost::asio::async_read(m_socket,
						m_read_buffer.prepare(protocol::P_BUFFER_SIZE),
						boost::asio::transfer_at_least(protocol::packet::HEADER_SIZE),
						boost::bind(&session::on_read,
							    this_ptr(),
							    boost::asio::placeholders::error,
							    boost::asio::placeholders::bytes_transferred));
		}
		else if((bytes_to_read != 0) &&
			((ec == error::header_incomplete) ||
			 (ec == error::packet_incomplete))) // header or packet incomplete
		{
			const std::size_t buffer_size = (std::max)(bytes_to_read,
								   m_read_buffer.left_space());
			boost::asio::async_read(m_socket,
						m_read_buffer.prepare(buffer_size),
						boost::asio::transfer_at_least(bytes_to_read),
						boost::bind(&session::on_read,
							    this_ptr(),
							    boost::asio::placeholders::error,
							    boost::asio::placeholders::bytes_transferred));
		}
		// 以下是对不可恢复错误的处理
		// 
		// if(ec == error::bad_magic) // 是magic错误，则关闭连接
		// or socket error
		// or other unknown error
		else
		{
			LOG_ENV(const boost::asio::const_buffer data = m_read_buffer.data());
			LOG_ENV(const std::size_t size =  boost::asio::buffer_size(data));

			LOG_IF_ELSE((ec != boost::asio::error::operation_aborted),
				    ERROR,
				    "read error, data size in read buffer = " << size
				    << ", bytes_transferred = " << bytes_transferred
				    << ", ec = " << ec << " - " << ec.message(),
				    TRACE,
				    "read operator cancelled");
			
			dispatch_error(ec);
			m_async_read_actived = false;
			close();
		}
	}

	void session::dispatch_error(const error_code &ec)
	{
		protocol::packet_ptr null_ptr;
		error_code e = ec; // make a local copy
		for(handler_map_type::iterator it = m_handler_map.begin();
		    (it != m_handler_map.end());
		    ++it)
		{
			LOG(ERROR, "dipatch handler for packet, id = " << it->first
			    << ", ec = " << ec << " - " << ec.message());
			this->get_upper_level_service().dispatch(boost::bind(it->second,
									     null_ptr, e));
		}
		m_handler_map.clear();
	}

	void session::dispatch(packet_ptr p, const error_code &ec)
	{
		const std::size_t id = p->get_sequence_number();
		const handler_map_type::iterator it = m_handler_map.find(id);
		const handler_map_type::iterator end = m_handler_map.end();
		
		if(it != end)
		{
			handler_type handler = it->second;
			m_handler_map.erase(it);
			// 向上层发起回调函数
			this->get_upper_level_service().dispatch(boost::bind(handler, p, ec));
		}
		LOG_IF((it == end),
		       ERROR,
		       "cannot found handler for packet = " << p->to_string());
	}

} // namespace xbase

